#!/usr/local/bin/perl -w
use strict;
use utf8;

use EV;
use HTTP::Parser::XS qw(parse_http_request);
use IO::Socket::INET;
use JSON::XS;
use Digest::MD5 qw(md5_hex);
require HTTP::Status;
use Time::HiRes qw(gettimeofday);

my $port = shift || 12345;
my $debug = shift;
binmode STDOUT,":utf8";

# список клиентов
my %client;
# список клиентов ожидающих сообщений
my @waiting;
# список входящих сообщений
my @messages;
# число удаленных сообщений из списка
my $shift = 0;
# все ожидающие чтения слушатели
my %readers;
# все ожидающие записи слушатели
my %writers;

my %command = (
	login => \&cmd_login,
	check => \&cmd_check,
	message => \&cmd_message,
);

# серверный сокет
my $lsn = new IO::Socket::INET(
	    Listen => 100,
	    LocalHost => "127.0.0.1",
	    LocalPort => $port,
	    ReuseAddr => 1,
	);

# слушаем основной сокет
my $w = EV::io $lsn, EV::READ, \&onstream;
EV::loop;

sub ts {
	return sprintf "%02d:%02d:%02d",(localtime)[2,1,0];
}
sub onstream {
	# принимаем подключение
	if($debug){
		print "accept\n";
	}
    my ($watcher, $revents) = @_;
    my $new = $lsn->accept;
    unless($new){
		if($debug){
			print "accept failed: $!\n";
		}
		return 0;
    }
    my $nw = EV::io $new, EV::READ, \&onread;
	# сохраняем слушателя, сокет и буфер с принятыми данными
    $readers{int $nw} = [$nw,$new,""];
}
sub onread{
	# Получили данные
	my ($w, $r) = @_;
	my $ary = $readers{int $w};
	my $buf;
	$ary->[1]->recv($buf,1024);
	if($debug){
		my $buf2 = $buf;
		#utf8::decode($buf2);
		print "read: $buf2\n";
	}
	if($buf){
		# есть данные, цепляем к предыдущим если есть
		unless(process($ary->[1], $buf = ($ary->[2] .= $buf))){
			# сообщение не полное, ждем дальше
			return 0;
		}
	}
	# раобота с содинением закончена, закрываем и чистим
	delete $readers{int $w};
	# сокет должен сам закрытся
}

sub process {
	my ($fh,$str) = @_;
	if(length($str) > 4096){
		# очень длинные запросы не обрабатываем
		send_message($fh,{status=>200,type=>"error",text=>"too long message",ts=>ts()});
		return 1;
	}
	my $param = {};
	my $header_size = parse_http_request($str,$param);
	if($header_size == -2){
		# неполное сообщение
		return undef;
	}elsif($header_size == -1){
		# поломанное сообщение
		send_message($fh,{status=>200,type=>"error",text=>"broken message",ts=>ts()});
		return 1;
	}
	my $body = substr($str,$header_size);
	if($param->{CONTENT_LENGTH} && $param->{CONTENT_LENGTH} > length($body)){
		return undef;
	}
	(my $cmd = $param->{PATH_INFO}) =~ s|.*/||;
	my $msg = decode_json($body);
	if(exists $command{$cmd}){
		$command{$cmd}->($fh,$msg);
		return 1;
	}else{
		send_message($fh,{status=>200,type=>"error",text=>"invalid command: $cmd",ts=>ts()});
		return 1;
	}
}
# отсылаем сообщение на сокет
# соообщение будет отослано как только это будет возможно
# сокет хранится только здесь
# сообщение в виде perl-структуры
sub send_message {
	my ($fh,$msg) = @_;
	my $status = delete $msg->{status} || 200;
	my $body = encode_json($msg);
	my $ret = "HTTP/1.1 ".$status." ".HTTP::Status::status_message($status)."\n";
	$ret .= "Content-Length: ".length($body)."\n";
	$ret .= "Content-Type: text/x-javascript\n";
	$ret .= "\n";
	$ret .= $body;
	my $nw = EV::io $fh, EV::WRITE, \&onwrite;
	# создаем слушателя и сохраняем необходимые данные
	$writers{int $nw} = [$nw,$fh,$ret,$msg->{type}];
	return undef;
}
sub onwrite {
	my ($w, $r) = @_;
	my $ary = $writers{int $w};
	my $need = length($ary->[2]);
	if($debug){
		my $buf = $ary->[2];
		#utf8::decode $buf;
		print "write: $buf\n";
	}
	my $rs = $ary->[1]->send($ary->[2]);
	if($rs){
		$need -= $rs;
		# если отправлено не все, то продолжаем ждать
		if($need > 0){
			substr($ary->[2],0,$rs,"");
			if($debug){
				print "write only $rs need $need\n";
			}
			return 0;
		}
		if($debug){
			print "write ok $rs $need $ary->[3]\n";
		}
	}
	delete $writers{int $w};
}
sub generate_sid {
	return md5_hex(gettimeofday(),rand());
}
# обработчики команд
sub cmd_login {
	my ($fh,$msg) = @_;
	unless($msg->{login}){
		send_message($fh,{status=>200,type=>"error",text=>"need login",ts=>ts()});
		return undef;
	}
	put_message({ts => ts(),text => "logged [$msg->{login}]"});

	my $sid = generate_sid();
	$client{$sid} = {
		login => $msg->{login},
		last => $shift + @messages,
	};
	send_message($fh,{status=>200,ts=>ts(),type=>"login",text=>"welcome [$msg->{login}]",sid => $sid});
}

sub cmd_message {
	my ($fh,$msg) = @_;
	if(my $cln = $client{$msg->{sid}||''}){
		send_message($fh,{status=>200,type=>"system",text=>"ok",ts=>ts()});
		put_message({ts=>ts(),from=>$cln->{login},text=>$msg->{text}});
	}else{
		send_message($fh,{status=>200,type=>"error",text=>"not authority",ts=>ts()});
	}
}
# вот здесь нужно ждать
sub cmd_check {
	my ($fh,$msg) = @_;
	my $cln;
	unless($cln = $client{$msg->{sid}||''}){
		send_message($fh,{status=>200,type=>"error",text=>"not authority",ts=>ts()});
	}
	# в список ожидающих поступления сообщений нужно сохранить сокет
	my $current = $shift + @messages;
	my $last = $cln->{last};
	# проверяем, есть ли новые сообщенияс последней отправки
	if($last < $current){
		# есть
		# отсылаем только последние 1000 сообщений
		if(@messages - ($last -= $shift) > 1000){
			$last = @messages - 100;
		}
		my @str;
		for(my $i = $last; $i < @messages; $i++){
			my $m = $messages[$i];
			push @str,{from=>$m->{from}||"",ts=>$m->{ts},text=>$m->{text}};
		}
		send_message($fh,{status=>200,type=>"list",list=>\@str});
		$cln->{last} = $shift + @messages;
	}else{
		# нет
		push @waiting,{fh => $fh, sid => $msg->{sid}};
	}
}
# расслка входящего сообщения
sub put_message {
	my ($msg) = @_;
	push @messages,$msg;
	if(@messages > 20000){
		splice @messages,0,5000;
		$shift += 5000;
	}
	# пройти по всем ожидающим и отправить
	my $current = $shift + @messages;
	# после этого сообщения уже никто не ждет и поэтому
	# они удаляются
	while(my $c = pop @waiting){
		my $cln = $client{$c->{sid}};
		my $last = $cln->{last};
		# отсылаем только последние 1000 сообщений
		if(@messages - ($last -= $shift) > 1000){
			$last = @messages - 100;
		}
		my @str;
		for(my $i = $last; $i < @messages; $i++){
			my $m = $messages[$i];
			push @str,{from=>$m->{from}||"",ts=>$m->{ts},text=>$m->{text}};
		}
		send_message($c->{fh},{status=>200,type=>"list",list=>\@str});
		$cln->{last} = $shift + @messages;
	}
}
